package com.gomro.galaxy.foundation.bi.engine.analysis;

import cn.gomro.cloud.core.common.utils.DateUtils;
import cn.gomro.cloud.core.common.utils.EncryptUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemReader;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemWriter;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.Constants;
import com.gomro.galaxy.foundation.bi.domain.event.VendorTradingVolumeEvent;
import com.gomro.galaxy.foundation.bi.domain.model.*;
import com.gomro.galaxy.foundation.bi.domain.model.analysis.ClientSettlementDataStatisticsVolume;
import com.gomro.galaxy.foundation.bi.domain.model.analysis.CustomerTradeVolume;
import com.gomro.galaxy.foundation.bi.domain.model.analysis.VendorTradingVolume;
import com.gomro.galaxy.foundation.bi.engine.AnalysisLastTime;
import com.gomro.galaxy.foundation.bi.engine.model.AnalysisCache;
import com.gomro.galaxy.foundation.bi.exception.BiException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.Sum;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.lang.NonNull;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;

import static com.gomro.galaxy.foundation.bi.engine.util.EngineUtils.MAX_BATCH_NUM;
//客户结算数据统计
/**
 * 公采结算数据JOB
 *
 * @author ZhangLei
 */
@Slf4j
@Configuration
public class ClientSettlementDataStatisticsVolumeJob {
    private static final String BiAnalysisLastName = Constants.BiDataMartIndexPrefix + AnalysisLastTime.TableName;
    private static final String BiWarehouseOrganizationName = Constants.BiWarehouseIndexPrefix + AcquireOrganization.TableName;
    private static final String BiWarehouseQuotationGoodsName = Constants.BiWarehouseIndexPrefix + AcquireQuotationGoods.TableName;
    private static final String BiWarehouseOrderGoodsName = Constants.BiWarehouseIndexPrefix + AcquireOrderGoods.TableName;
    private static final String BiWarehouseVendorOrderName = Constants.BiWarehouseIndexPrefix + AcquireVendorOrder.TableName;
    private static final String BiWarehouseOrderGoodsRefundName = Constants.BiWarehouseIndexPrefix + AcquireOrderGoodsRefund.TableName;
    private static final String BiDataMartName = Constants.BiDataMartIndexPrefix + ClientSettlementDataStatisticsVolume.TableName;
    private static final String JobName = Constants.JobBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName;
    private static final String StepName1 = Constants.StepBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 1;
    private static final String StepName2 = Constants.StepBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 2;
    private static final String StepName3 = Constants.StepBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 3;
    private static final String StepName4 = Constants.StepBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 4;
    private static final String StepName5 = Constants.StepBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 5;

    private static final String ReaderName1 = Constants.ReaderBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 1;
    private static final String ReaderName2 = Constants.ReaderBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 2;
    private static final String ReaderName3 = Constants.ReaderBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 3;
    private static final String ReaderName4 = Constants.ReaderBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 4;
    private static final String ReaderName5 = Constants.ReaderBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 5;
    private static final String WriterName1 = Constants.WriterBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 1;
    private static final String WriterName2 = Constants.WriterBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 2;
    private static final String WriterName3 = Constants.WriterBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 3;
    private static final String WriterName4 = Constants.WriterBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 4;
    private static final String WriterName5 = Constants.WriterBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 5;
    private static final String ProcessName1 = Constants.ProcessBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 1;
    private static final String ProcessName2 = Constants.ProcessBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 2;
    private static final String ProcessName3 = Constants.ProcessBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 3;
    private static final String ProcessName4 = Constants.ProcessBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 4;
    private static final String ProcessName5 = Constants.ProcessBeanNamePrefix + ClientSettlementDataStatisticsVolume.TableName + 5;

    private ApplicationEventPublisher applicationEventPublisher;
    private BusProperties busProperties;
    private JobBuilderFactory jobBuilderFactory;
    private StepBuilderFactory stepBuilderFactory;
    private ElasticSearchRest rest;
    private RestHighLevelClient client;
    private ObjectMapper mapper;
    private AnalysisCache<CustomerTradeVolume> cache = new AnalysisCache();

    public ClientSettlementDataStatisticsVolumeJob(ApplicationEventPublisher applicationEventPublisher, BusProperties busProperties, JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, RestHighLevelClient client, ObjectMapper mapper) {
        this.applicationEventPublisher = applicationEventPublisher;
        this.busProperties = busProperties;
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.client = client;
        this.mapper = mapper;
        this.rest = new ElasticSearchRest(client, mapper);
    }


    private JobExecutionListener jobExecutionListener = new JobExecutionListener() {

        @Override
        public void beforeJob(@NonNull JobExecution jobExecution) {

            ElasticSearchRest rest = new ElasticSearchRest(client, mapper);

            // 初始化数据索引
            if (!rest.indexExists(BiDataMartName)) {

                rest.indexCreate(BiDataMartName);
                rest.indexMappingPut(BiDataMartName, "analysis/" + BiDataMartName + ".json", true);
                rest.indexFlush(BiDataMartName, null, null);
            }

            // 初始化最后更新时间索引
            if (!rest.indexExists(BiAnalysisLastName)) {

                rest.indexCreate(BiAnalysisLastName);
                rest.indexMappingPut(BiAnalysisLastName, "analysis/" + BiAnalysisLastName + ".json", true);
                rest.indexFlush(BiAnalysisLastName, null, null);
            }
        }

        @Override
        public void afterJob(@NonNull JobExecution jobExecution) {
            if (Objects.nonNull(cache.getCache())) {
                cache.getCache().clear();
            }
            // 回写最后更新时间
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
                log.debug("last time ---> {}", DateUtils.dateToStr(cache.getLast()));
                rest.documentCreate(BiAnalysisLastName, VendorTradingVolume.TableName, AnalysisLastTime.builder().id(VendorTradingVolume.TableName).last(cache.getLast()).build());
                applicationEventPublisher.publishEvent(new VendorTradingVolumeEvent(cache.getLast()));
            }
        }
    };


}
